package com.example.flinktest.transformation;

import com.example.flinktest.source.Event;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransMapTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //数据源
        DataStreamSource<Event> streamSource = env.fromElements(
                new Event("junsong", "./home", 100L),
                new Event("boy", "./cart", 200L)
        );
        //转换算子，业务处理部分

        //传入匿名类，实现MapFunction
        streamSource.map(new MapFunction<Event, Object>() {
            @Override
            public Object map(Event event) throws Exception {
                return event.user;
            }
        });
        //打印结果
        //传入MapFunction的实现类
        streamSource.map(new UserExtractor()).print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static class UserExtractor implements MapFunction<Event, String> {

        @Override
        public String map(Event event) throws Exception {
            return event.user;
        }
    }
}
